package com.youxuit.youxu.kecheng.CASandAtomic;

import cn.hutool.core.collection.CollUtil;

import java.util.List;
import java.util.concurrent.*;

/**
 * DataPush
 * <p>
 * 异步推送数据
 * 整体步骤：
 * 1、在消息生产者MessageProduceToSync中，使用多线程快速生产1000W条信息，考虑到内存问题，只使用了int类型来模拟每条数据。
 * 2、将生产的消息异步推送到消息消费者-->MessageConsumer的阻塞队列中等待消费。
 * 3、启动另一个线程模拟消息消费者（MessageConsumer），该消费者中使用一个阻塞队列来接收消息，
 * 阻塞队列中有消息时消费，无消息时则阻塞。
 * 4、生产者生产完1000w条消息后，关闭产生的线程，释放资源。
 *
 * @author ZhenXu
 * @date 2023-12-20 21:22
 **/
public class MessageProduceToSync {

    /**
     * ConcurrentLinkedQueue:待推送数据集合
     */
    public static final ConcurrentLinkedQueue<Integer> MOCK_MSG_LINKED_QUEUE = new ConcurrentLinkedQueue<>();

    /**
     * CopyOnWriteArrayList:待推送数据集合
     */
//    public static final CopyOnWriteArrayList<Integer> MOCK_MSG_LINKED_QUEUE = new CopyOnWriteArrayList<>();

    /**
     * 线程设置数，当前系统CPU核数/2
     */
    private static final Integer MAX_CPU_CORES = Runtime.getRuntime().availableProcessors() / 2;

    /**
     * 初始化线程池
     */
    public static final ThreadPoolExecutor THREAD_POOL_EXECUTOR = buildThreadPoolExecutor();

    public static void main(String[] args) throws InterruptedException {

        //初始化系统开始时间和结束时间，用来计算程序执行的耗时
        long startTime;
        long endTime;

        //初始化线程计数器
        int firstLoopCountNum = 30;
        CountDownLatch countDownLatch = new CountDownLatch(firstLoopCountNum);

        startTime = System.currentTimeMillis();

        //异步创建消息
        syncCreateMsg(firstLoopCountNum, countDownLatch);

        //计算生产消息的耗时
        endTime = System.currentTimeMillis();
        System.out.println("----添加元素共花费: " + (endTime - startTime) + " 毫秒");
        System.out.println(MOCK_MSG_LINKED_QUEUE.size());

        //创建消息消费者，并启动其线程开始监听生产者的消息推送
        MessageConsumer messageConsumer = new MessageConsumer();
        messageConsumer.start();

        startTime = System.currentTimeMillis();

        //异步推送消息
        syncPushMsg();

        endTime = System.currentTimeMillis();
        System.out.println("----推送数据共花费: " + (endTime - startTime) + " 毫秒");

        // 关闭线程池，这会等待所有任务执行完毕
        THREAD_POOL_EXECUTOR.shutdown();
        boolean isStop = THREAD_POOL_EXECUTOR.awaitTermination(1, TimeUnit.HOURS);

//        messageConsumer.threadFlag = false;
    }


    /**
     * 说明: 异步创建消息
     * 最少创建100W条
     * <p>
     *
     * @param firstLoopCountNum 外循环与计数器的值
     * @param countDownLatch    计数器
     * @throws InterruptedException 可针对异常情况进行消费补偿
     * @author ZhenXu
     * @date 2023-12-22 12:27
     */
    private static void syncCreateMsg(int firstLoopCountNum, CountDownLatch countDownLatch) throws InterruptedException {
        //需要循环的次数，通过countNum*loopCountNum获取共需要生产多少条消息
        int secondLoopCountNum = 1000000;

        //疑惑：在数据量只有30W时，这里使用异步线程更慢，同步只要0.8秒左右，异步却要1秒左右，为什么呢？
        //循环生产消息
        for (int i = 1; i <= firstLoopCountNum; i++) {
            int finalI = i;
            THREAD_POOL_EXECUTOR.submit(() -> {
                try {
                    for (int j = 1; j <= secondLoopCountNum; j++) {
                        //计算消息的值
                        int numMsg = finalI * j;

                        //打印也很耗费性能
//                        System.out.println(num);

                        //将消息存入队列中
                        MOCK_MSG_LINKED_QUEUE.add(numMsg);
                    }
                } finally {
                    countDownLatch.countDown();
                }
            });
        }

        //等待消息创建完毕后再往后执行
        countDownLatch.await();
    }

    /**
     * 说明: 异步推送消息
     * <p>
     *
     * @throws InterruptedException 可针对异常情况进行消费补偿
     * @author ZhenXu
     * @date 2023-12-22 12:27
     */
    private static void syncPushMsg() throws InterruptedException {
        //重新计算计数器的值，该计数器用来控制异步推送消息的次数是否准确。
        int firstLoopCountNum = MOCK_MSG_LINKED_QUEUE.size() / 10000;

        //异步情况下，1000W的数据大概35秒推送完成，同步则需要23秒，反而更快。是因为打印的原因导致的吗？
        CountDownLatch msgCountDownLatch = new CountDownLatch(firstLoopCountNum);

        //循环推送数据
        for (List<Integer> msgs : CollUtil.split(MOCK_MSG_LINKED_QUEUE, 10000)) {
            //异步线程推送消息，在3千万的数据上才能明显感到异步的效率提升
            THREAD_POOL_EXECUTOR.submit(() -> {
                msgs.forEach(msg -> {
                    try {
                        MessageConsumer.pushMsg(msg);
                    } catch (InterruptedException e) {
                        //若消息推送异常，可以在此处进行补偿，进行入库记录或者推入其它补偿队列都可以。
                        // errorPushMessage.insert(msg);
                        throw new RuntimeException(e);
                    }
                });

                msgCountDownLatch.countDown();
            });
        }

        msgCountDownLatch.await();
    }

    /**
     * 说明: 线程池配置
     * 核心线程数与最大线程数都为当前系统的CPU最大核数
     *
     * @return {@link ThreadPoolExecutor}
     * @author ZhenXu
     * @date 2023-12-21 15:00
     */
    private static ThreadPoolExecutor buildThreadPoolExecutor() {
        // 创建线程池
        return new ThreadPoolExecutor(MAX_CPU_CORES,
                MAX_CPU_CORES,
                1,
                TimeUnit.SECONDS,
                new LinkedBlockingQueue<>(100000),  // 阻塞队列
                new MyCustomThreadFactory(),
                new CustomRejectedExecutionHandler() // 线程拒绝策略
        ) {

            @Override
            protected void afterExecute(Runnable r, Throwable t) {
//                System.out.println("任务执行完成: " + r);
//                System.out.println(t.getMessage());
                super.afterExecute(r, t);
            }

            @Override
            protected void terminated() {
                System.out.println("线程池关闭，线程回收完成");
                super.terminated();
            }
        };
    }
}

